package ar.edu.itba.pod.legajo45129.agentSimulation;

import java.io.Serializable;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.agent.runner.TargetedEvent;
import ar.edu.itba.pod.legajo45129.manager.NodeManager;
import ar.edu.itba.pod.multithread.EventDispatcher;

import com.google.common.base.Function;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Multimaps;
import com.google.common.collect.SetMultimap;

@SuppressWarnings("rawtypes")
public class DistributedMultithreadEventDispatcher implements EventDispatcher {
	// private RemoteEventDispatcher remoteEventDispatcher;

	protected static final int QUEUE_SIZE = 256;

	private static final double PROBABILITY = 0.8;
	private static final double THREHOLD = 0.3;
	private static Logger logger = Logger
			.getLogger(DistributedMultithreadEventDispatcher.class);

	private SetMultimap<Class, Agent> listeners = Multimaps
			.synchronizedSetMultimap(HashMultimap.<Class, Agent> create());
	private Map<Agent, BlockingQueue<Object>> queues = new MapMaker()
			.makeComputingMap(new Function<Agent, BlockingQueue<Object>>() {
				@Override
				public BlockingQueue<Object> apply(Agent input) {
					return new LinkedBlockingQueue<Object>(QUEUE_SIZE);
				}
			});

	private BlockingQueue<EventInformation> eventsQueue = new LinkedBlockingQueue<EventInformation>();
	private Thread pollingForEventInformation;

	private static final long WAIT_FOR_POLLING = 2000;

	public DistributedMultithreadEventDispatcher() {
		this.pollingForEventInformation = new Thread() {
			@Override
			public void run() {
				Random random = new Random();
				while (true) {
					try {
						sleep(WAIT_FOR_POLLING);

						EventInformation ei = eventsQueue.take();
						localPublish(ei.source(), ei.event());
						double r = random.nextDouble();
						Set<NodeInformation> nodes = NodeManager.getInstance()
								.getClusterAdministration().connectedNodes();
						for (NodeInformation n : nodes) {
							RemoteEventDispatcher nodeRemoteEventDispatcher = NodeManager
									.getInstance().getRemoteEventDispatcher(
											n.host(), n.port());
							if (r > THREHOLD) {
								r *= nodeRemoteEventDispatcher.publish(ei) ? 1.0
										: PROBABILITY;
							}
							// nodeRemoteEventDispatcher.publish(ei);
						}

					} catch (InterruptedException e) {
						logger.warn(e);
					} catch (RemoteException e) {
						logger.warn(e);
					} catch (NotBoundException e) {
						logger.warn(e);
					}
				}
			}
		};
		this.pollingForEventInformation.start();
	}

	@Override
	public void register(Agent agent, Class<? extends Serializable> type) {
		listeners.put(type, agent);
	}

	@Override
	public void deregister(Agent agent, Class<? extends Serializable> type) {
		listeners.remove(type, agent);

	}

	/**
	 * Unregister an agent
	 * 
	 * @param agent
	 * @return the pending events
	 */
	public BlockingQueue<Object> deregister(Agent agent) {
		Set<Entry<Class, Agent>> entries = listeners.entries();

		synchronized (listeners) {
			Iterator<Entry<Class, Agent>> entriesIter = entries.iterator();
			while (entriesIter.hasNext()) {
				Entry<Class, Agent> entry = entriesIter.next();
				if (agent.equals(entry.getValue())) {
					entriesIter.remove();
				}
			}
			BlockingQueue<Object> pendingEvents = queues.get(agent);
			queues.remove(agent);
			return pendingEvents;
		}
	}

	@Override
	public void publish(Agent source, Serializable event)
			throws InterruptedException {
		EventInformation ei = new EventInformation(event, NodeManager
				.getInstance().getNodeInformation().id(), source);
		ei.setReceivedTime(System.currentTimeMillis());
		eventsQueue.put(ei);
	}

	public void localPublish(Agent source, Serializable event)
			throws InterruptedException {
		String target = (event instanceof TargetedEvent) ? ((TargetedEvent) event)
				.target() : null;
		Set<Agent> agents = listeners.get(event.getClass());
		for (Agent agent : agents) {
			if (target == null || target.equals(agent.name())) {
				BlockingQueue<Object> queue = queues.get(agent);
				queue.put(event);
			}
		}
	}

	/**
	 * Returns the next event on the event queue, or waits for a set amount of
	 * time for an event to appear. If no event appears after timeout, it will
	 * be
	 * 
	 * @throws InterruptedException
	 *             if there is an interruption while waiting
	 */
	@SuppressWarnings("unchecked")
	@Override
	public <T extends Serializable> T waitFor(Agent agent, Class<T> baseType)
			throws InterruptedException {
		return (T) queues.get(agent).take();
	}

	/**
	 * Sets a predefined events queue for an agent
	 * 
	 * @param agent
	 * @param queue
	 */
	public void setAgentQueue(Agent agent, BlockingQueue<Object> queue) {
		queues.put(agent, queue);
	}

}
